package org.yonggan.page;

import com.alibaba.fastjson.JSONObject;
import org.apache.spark.SparkConf;
import org.apache.spark.api.java.JavaRDD;
import org.apache.spark.api.java.JavaSparkContext;
import org.apache.spark.sql.Row;
import org.apache.spark.sql.SQLContext;
import org.yonggan.constant.Constants;
import org.yonggan.dao.ITaskDAO;
import org.yonggan.dao.factory.DAOFactory;
import org.yonggan.domain.Task;
import org.yonggan.util.ParamUtils;
import org.yonggan.util.SparkUtils;

import java.util.Date;

/**
 * 页面单跳转换率
 *
 * 基本的需求：

     1、接收J2EE系统传入进来的taskid，从mysql查询任务的参数，日期范围、页面流id
     2、针对指定范围日期内的用户访问行为数据，去判断和计算，页面流id中，每两个页面组成的页面切片，它的访问量是多少
     3、根据指定页面流中各个页面切片的访问量，计算出来各个页面切片的转化率
     4、计算出来的转化率，写入mysql数据库中
 *
 */
public class PageOneStepConvertRateSpark {


    public static void main(String[] args) {

        // 构造上下文
        SparkConf conf = new SparkConf();
        conf.setAppName("页面单跳转换率");
        SparkUtils.setMaster(conf);

        JavaSparkContext sc = new JavaSparkContext(conf);
        // SQLContext
        SQLContext sqlContext = SparkUtils.getSQLContext(sc.sc());
        // 生成模拟数据
        SparkUtils.makeMock(sc,sqlContext);

        // 查询任务
        Long taskId = ParamUtils.getTaskIdFromArgs(args, Constants.SPARK_LOCAL_TASKID_PAGE);
        ITaskDAO taskDAO = DAOFactory.getTaskDAO();
        // 任务
        Task task = taskDAO.findById(taskId);
        if (task == null) {
            System.out.println(new Date() + ": cannot find this task with id [" + taskId + "].");
            return;
        }
        // 任务参数
        JSONObject taskParam = JSONObject.parseObject(task.getTaskParam());


        // 查询指定日期的行为数据数据
        JavaRDD<Row> actionRDDByDateRange = SparkUtils.getActionRDDByDateRange(sqlContext, taskParam);



    }

}
